package lagou.boot;

import com.lagou.serializer.JSONSerializer;
import com.lagou.serializer.RpcDecoder;
import com.lagou.serializer.RpcEncoder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lagou.handler.UserClientHandler;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author zy
 * @version 1.0
 * @date 2021/2/4 19:42
 */
public class Mapper implements Watcher {

    private static Map<String, UserClientHandler> userClientHandlerMap = new HashMap<String, UserClientHandler>();

    private static CuratorFramework client;

    public  CuratorFramework getClient() {
        return client;
    }

    public  void setClient(CuratorFramework client) {
        Mapper.client = client;
    }

    private List<String> children;

    public  Map<String, UserClientHandler> getUserClientHandlerMap() {
        return userClientHandlerMap;
    }

    public  void setUserClientHandlerMap(Map<String, UserClientHandler> userClientHandlerMap) {
        Mapper.userClientHandlerMap = userClientHandlerMap;
    }

    public static ExecutorService getExecutorService() {
        return executorService;
    }

    public static void setExecutorService(ExecutorService executorService) {
        Mapper.executorService = executorService;
    }

    //1.创建一个线程池对象  -- 它要处理我们自定义事件
    private static ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    //3.编写方法,初始化客户端  ( 创建连接池  bootStrap  设置bootstrap  连接服务器)
    public  void initClient(String ip, int port) throws Exception {
        //1) 初始化UserClientHandler
        final UserClientHandler userClientHandler  = new UserClientHandler();
        //2)创建连接池对象
        EventLoopGroup group = new NioEventLoopGroup();
        //3)创建客户端的引导对象
        Bootstrap bootstrap =  new Bootstrap();
        //4)配置启动引导对象
        bootstrap.group(group)
                //设置通道为NIO
                .channel(NioSocketChannel.class)
                //设置请求协议为TCP
                .option(ChannelOption.TCP_NODELAY,true)
                //监听channel 并初始化
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //获取ChannelPipeline
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        //设置编码
                        pipeline.addLast(new RpcEncoder(Object.class, new JSONSerializer()));
                        pipeline.addLast(new RpcDecoder(Object.class, new JSONSerializer()));
//                        pipeline.addLast(new StringEncoder());
//                        pipeline.addLast(new StringDecoder());
                        //添加自定义事件处理器
                        pipeline.addLast(userClientHandler);
                    }
                });

        //5)连接服务端
        bootstrap.connect(ip,port).sync();
        userClientHandlerMap.put(ip+":"+port,userClientHandler);

    }

    public Mapper() throws Exception {
        // 从Zookeeper中获取到所有服务端的ip和端口信息，然后建立连接
        // 定义重载策略
        RetryPolicy exponentialBackoffRetry = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                // 连接地址
                .connectString("121.4.48.59:2181")
                // 会话超时时间
                .sessionTimeoutMs(50000)
                // 连接超时时间
                .connectionTimeoutMs(30000)
                // 重载策略
                .retryPolicy(exponentialBackoffRetry)
                // 独立命名空间
                .namespace("netty")
                .build();

        client.start();
        /* 获取初始节点列表 */
        children = client.getChildren().usingWatcher(this).forPath("/");
        System.out.println("初始监听目录节点列表：" + children);
    }



    @Override
    public void process(WatchedEvent event) {
        try {
            /* 获取最新节点列表 */
            final List<String> newChildren = client.getChildren().usingWatcher(this).forPath("/");

            /* 新节点列表大于旧节点列表，说明有节点上线，反之有节点下线 */
            if (newChildren.size() > children.size()) {
                for (String child : newChildren) {
                    /* 取出旧节点列表中不包含新节点，即为上线 */
                    if (!children.contains(child)) {
                        System.out.println("上线节点：" + child);
                        ConsumerBoot.start();
                    }
                }
            } else {
                for (String child : children) {
                    /* 取出新节点列表中不包含旧节点，即为下线 */
                    if (!newChildren.contains(child)) {
                        System.out.println("下线节点：" + child);
                    }
                }
            }

            /* 更新节点列表 */
            children = newChildren;
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
